package com.gt.stream.req

import com.gt.stream.req.util.{JDBCUtil, MyUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}

import java.sql.{Connection, PreparedStatement, ResultSet}
import java.text.SimpleDateFormat
import java.util.Date
import scala.collection.mutable.ListBuffer


object SparkStreaming_req_001 {

  def main(args: Array[String]): Unit = {
    //1.
    val conf: SparkConf = new SparkConf().setAppName("xx").setMaster("local[1]")
    val ssc = new StreamingContext(conf, Seconds(1))

    //2.
    val kafkaParams: Map[String, Object] = MyUtil.createKafkaParam()

    //3.
    val stream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
      ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](Set("test"), kafkaParams)
    )

    //4.
    val data: DStream[String] = stream.map(_.value())

    //5.
    val dataWindow: DStream[String] = data.window(Seconds(5), Seconds(5))
    //6. 转换为ds
    val adClickData: DStream[AdClickData] = dataWindow.map(line => {
      val arr: Array[String] = line.split(" ")
      AdClickData(arr(0), arr(1), arr(2), arr(3), arr(4))
    })


    val clickData: DStream[((String, String, String), Int)] = adClickData.transform(rdd => {
      //TODO 周期性获取黑名单数据
      val blackList = new ListBuffer[String]()
      val connection: Connection = JDBCUtil.getConnection()
      val preparedStatement: PreparedStatement = connection.prepareStatement("select * from black_list")
      val rs: ResultSet = preparedStatement.executeQuery()
      while (rs.next()) {
        val userId: String = rs.getString(1)
        blackList.append(userId)
      }
      rs.close()
      preparedStatement.close()
      connection.close()

      val sdf = new SimpleDateFormat("yyyy-MM-dd")
      rdd.filter(adClickData => {
        //TODO 过滤黑名单数据
        !blackList.contains(adClickData.userid)
      }).map(data => {
        val day = sdf.format(new Date(data.ts.toLong))
        ((day, data.userid, data.adid), 1)
      }).reduceByKey(_ + _)
    })

    //TODO 统计点击数据并更新到 ad点击信息表
    clickData.foreachRDD(rdd => {
      rdd.foreach(data =>{
        val count  = data._2
        if(count >= 30){
          //超过30次直接加入黑名单

        }else{
          //单批次没有超过30次，查询数据库，并累加
          //更新数据到数据库
          //判断是否超过30次，超过则添加到黑名单

        }
      })
    })



    //TODO 判断点击数量是否超过阈值，超过，则更新黑名单表



    //x.
    ssc.start()
    ssc.awaitTermination()
  }

  //广告点击数据
  case class AdClickData(ts: String, area: String, city: String, userid: String, adid: String)

}
